<?php

/**
 * Created by IntelliJ IDEA.
 * User: ty42
 * Date: 4/20/16
 * Time: 5:10 PM
 *
 * 1. 拆分可执行任务, 写数据库, 写队列
 * 2. 读待执行任务队列, 执行任务. 如果是群发类型, 写队列
 * 3. 执行细分任务
 */
class Msgtaskm extends CI_Model
{
    /**
     * 表名
     */
    const TBL_MSG_TASK              = 'ml_msg_task';
    const TBL_MSG_RUN               = 'ml_msg_task_run';

    /**
     * msn 实体
     */
    public $mns = null;

    /**
     * 队列名称
     */
    const QNAME_MSG_ATOM            = 'qy-msg-atom';

    /**
     * 任务状态
     */
    const TSK_STAT_READY            = 0;
    const TSK_STAT_ING              = 1;
    const TSK_STAT_DONE             = 2;


    /**
     * 支持的发送类型
     */
    const MT_METHOD_UNDEFINED       = 0;        // 未定义
    const MT_METHOD_PM              = 1;        // 私信, private msg


    /**
     * 发送目标类型
     */
    const TGT_TYP_IDS               = 0;        // 按 id
    const TGT_TYP_GROUP             = 1;        // 按群组、科室
    const TGT_TYP_ALL               = 2;        // 全部用户


    /**
     * 发送业务子类型, 私信
     */
    const SUB_TYP_PM_RY             = 1;        // 融云 IM
    const SUB_TYP_PM_JMESSAGE       = 2;        // 极光 IM


    /**
     * 目标 id 分隔符
     */
    const CHAR_SEP = ';;';


    /**
     * 发送分组, 组员上限
     */
    const LMT_GROUP                 = 1000;


    static $sub_method_arr = array(
        self::MT_METHOD_PM => array(
            self::SUB_TYP_PM_RY,
            self::SUB_TYP_PM_JMESSAGE,
        ),
    );

    /**
     * 发送方式配置
     */
    static $sub_method_conf = array(
        self::SUB_TYP_PM_RY => array(
            'is_support_all'=> false,
            'is_support_group'=> false,
            'is_support_batch_ids'=> true,
            'limit_batch_ids'=> 1000,
            'interval'=> 12,
            'cond_group_select' => array("softversion >= " => "2.0.0"),
        ),
        self::SUB_TYP_PM_JMESSAGE => array(
            'is_support_all'=> false,
            'is_support_group'=> false,
            'is_support_batch_ids'=> false,
            'limit_batch_group'=> 0,
            'limit_batch_ids'=> 1000,
            'interval'=> 15,
            'cond_group_select' => array("softversion > " => 0, "softversion<" => "2.0.0"),
        ),
    );


    public function __construct()
    {
        parent::__construct();

        $this->load->helper('fllogs');
    }


    /**
     * @return Mnslib
     */
    protected function get_instance_of_mnslib()
    {
        if (!$this->mns) {
            if (!class_exists('Mnslib')) {
                include_once APPPATH . 'libraries/Mnslib.php';
            }

            $this->mns = new Mnslib();
        }

        return $this->mns;
    }



    /**
     * 获取可执行任务
     *
     * @return array
     */
    public function prepared_tasks()
    {
        $whe = array("stat"=> self::TSK_STAT_READY, "stime<="=> time());
        $query = $this->db->where($whe)->order_by("stime", "desc")->get(self::TBL_MSG_TASK);

        $result = array();
        if ($query) {
            $result = $query->result_array();
        }
        return $result;
    }


    /**
     * 更新任务状态
     *
     * @param $task_id
     * @param $status
     */
    public function update_task_status($task_id, $status)
    {
        $this->db->set("stat", $status);
        $this->db->set("utime", time());
        $this->db->where("mt_id", $task_id);
        $this->db->update(self::TBL_MSG_TASK);
    }


    public function update_msg_run($run_id, $data)
    {
        foreach ($data as $k => $v) {
            $this->db->set($k, $v);
        }
        $this->db->set("rutime", time());

        $this->db->where('run_id', $run_id);
        $this->db->update(self::TBL_MSG_RUN);
    }


    /**
     * 获取子发送类型列表
     *
     * @param $msg_task
     * @return array
     */
    public function get_sub_typ_arr($msg_task)
    {
        $sub_typ_lst = array();

        if ($msg_task["sub_typ"]) {
            // 发送时指定了子发送方法
            $sub_typ_lst = array(0=> $msg_task["sub_typ"]);
        }
        elseif ($msg_task["method"] == self::MT_METHOD_PM) {
            $sub_typ_lst = array(
                self::SUB_TYP_PM_RY,
                self::SUB_TYP_PM_JMESSAGE
            );
        }
        return $sub_typ_lst;
    }


    /**
     * 获取目标分组细分
     *
     * @param $msg_task
     * @return array
     */
    public function get_dest_arr($msg_task)
    {
        $dest_arr = array();
        if ($msg_task["method"] == self::MT_METHOD_PM) {
            if ($msg_task["typ"] == self::TGT_TYP_GROUP) {
                // 私信按群组发送, 需要拆分
                $dest_arr = explode(self::CHAR_SEP, $msg_task["to"]);
            }
            else {
                $dest_arr = array($msg_task["to"]);
            }
        }
        return $dest_arr;
    }


    public function subdivideUidToUids($uids_arr, $page_size=self::LMT_GROUP)
    {
        if (empty($uids_arr)) {
            return null;
        }

        $result = array();
        $pn = intval((count($uids_arr) - 1) / $page_size) + 1;
        for ($page = 0; $page < $pn; $page++) {
            $to_ids = array_slice($uids_arr, $page * $page_size, $page_size);

            $to_ids_str = implode(self::CHAR_SEP, $to_ids);

            $result[] = array(self::TGT_TYP_IDS, $to_ids_str);
        }
        return $result;
    }


    public function subdivideThreadToUids($thread_arr, $sub_method_conf, $page, $page_size)
    {
        $this->db->select("id");
        $this->db->where("is_show", "1");
        if (isset($sub_method_conf["cond_group_select"])) {
            foreach ($sub_method_conf["cond_group_select"] as $k => $v) {
                $this->db->where($k, $v);
            }
        }
        if (!empty($thread_arr)) {
            $this->db->where_in("thread", $thread_arr);
        }
        else {
            // 取全部用户
        }
        $query = $this->db->get('ml_fellow', $page_size, $page_size * $page);
        if (! $query) {
            return array();
        }
//        echo $this->db->last_query(), "\n";

        $dt = $query->result_array();
        $to_ids = array();
        foreach ($dt as $item) {
            $to_ids[] = $item["id"];
        }
        $to_ids_str = implode(self::CHAR_SEP, $to_ids);

        return array(self::TGT_TYP_IDS, $to_ids_str);
    }


    public function subdivideSendArr($sub_typ_conf, $tgt_typ, $to, $page, $page_size)
    {
        if ($tgt_typ != self::TGT_TYP_ALL && !$to) {
            return array();
        }

        if ($tgt_typ == self::TGT_TYP_GROUP) {
            if ($sub_typ_conf["is_support_group"] && $page == 0) {
                return array(self::TGT_TYP_GROUP, $to);
            }
            elseif (! $sub_typ_conf["is_support_group"]) {
                $sub_arr = explode(self::CHAR_SEP, $to);
                if ($sub_arr) {
                    return $this->subdivideThreadToUids($sub_arr, $sub_typ_conf, $page, $page_size);
                }
            }
        }
        elseif ($tgt_typ == self::TGT_TYP_IDS) {
            $uid_arr = explode(self::CHAR_SEP, $to);

            $to_ids = array_slice($uid_arr, $page * $page_size, $page_size);

            $to_ids_str = implode(self::CHAR_SEP, $to_ids);

            return array(self::TGT_TYP_IDS, $to_ids_str);
        }
        elseif ($tgt_typ == self::TGT_TYP_ALL) {
            if ($sub_typ_conf["is_support_all"] && $page == 0) {
                return array(self::TGT_TYP_ALL, '');
            }
            else {
                return $this->subdivideThreadToUids(null, $sub_typ_conf, $page, $page_size);
            }
        }
        return array();
    }


    /**
     * 按科室id取符合条件的用户id总数
     * @param $thread_ids
     * @param $sub_method_conf
     * @return int
     */
    public function getNumSubdivideThreadToUids($thread_ids, $sub_method_conf)
    {
        $this->db->where("is_show", "1");
        if (isset($sub_method_conf["cond_group_select"])) {
            foreach ($sub_method_conf["cond_group_select"] as $k => $v) {
                $this->db->where($k, $v);
            }
        }
        if (!empty($thread_ids)) {
            $this->db->where_in("thread", $thread_ids);
        }
        else {
            // 取全部用户
        }
        $nall = $this->db->count_all_results('ml_fellow');

        return $nall;
    }


    public function getNumSubdivideArr($sub_typ_conf, $tgt_typ, $to)
    {
        $nall = 0;
        if ($tgt_typ != self::TGT_TYP_ALL && !$to) {
            return $nall;
        }

        if ($tgt_typ == self::TGT_TYP_GROUP) {
            if ($sub_typ_conf["is_support_group"]) {
                $nall = 1;
            }
            else {
                $thread_ids = explode(self::CHAR_SEP, $to);
                if ($thread_ids) {
                    $nall = $this->getNumSubdivideThreadToUids($thread_ids, $sub_typ_conf);
                }
            }
        }
        elseif ($tgt_typ == self::TGT_TYP_IDS) {
            $uid_arr = explode(self::CHAR_SEP, $to);

            $nall = count($uid_arr);
        }
        elseif ($tgt_typ == self::TGT_TYP_ALL) {
            if ($sub_typ_conf["is_support_all"]) {
                $nall = 1;
            }
            else {
                $nall = $this->getNumSubdivideThreadToUids(null, $sub_typ_conf);
            }
        }

        return $nall;
    }

    
    /**
     * 将消息发送任务划分为可执行的子任务, 并写任务队列
     *
     * @param $msg_task
     * @return bool
     */
    public function subdivide_task($msg_task)
    {
        $msg_task_now = $this->db->select('stat')->where('mt_id', $msg_task["mt_id"])->get(self::TBL_MSG_TASK)->row_array();
        if ($msg_task_now["stat"] != self::TSK_STAT_READY) {
            # 任务已经开始执行
            return false;
        }

        $this->update_task_status($msg_task["mt_id"], self::TSK_STAT_ING);

        $mt_method = $msg_task["method"];
        # 获取发送子方式细分列表
        $sub_typ_arr = isset(self::$sub_method_arr[$mt_method]) ? self::$sub_method_arr[$mt_method] : null;
        if (empty($sub_typ_arr)) {
            return true;
        }
        foreach ($sub_typ_arr as $sub_typ) {
            $sub_typ_conf = isset(self::$sub_method_conf[$sub_typ]) ? self::$sub_method_conf[$sub_typ] : null;
            if (empty($sub_typ_conf)) {
                continue;
            }

            $nall = $this->getNumSubdivideArr($sub_typ_conf, $msg_task["typ"], $msg_task["to"]);

            if (! $nall) {
                continue;
            }

            $page_size = isset($sub_method_conf["limit_batch_ids"]) ? $sub_method_conf["limit_batch_ids"] : self::LMT_GROUP;

            $delay_seconds = 0;
            for ($page=0; $page * $page_size<$nall; $page++) {
                $tgt_arr = $this->subdivideSendArr($sub_typ_conf, $msg_task["typ"], $msg_task["to"], $page, $page_size);

                if (empty($tgt_arr)) {
                    continue;
                }

                $data = array(
                    "rt_id" => $msg_task["mt_id"],
                    "target_group_typ" => $tgt_arr[0],
                    "rsub_typ" => $sub_typ,
                    "to" => $tgt_arr[1],
                    "ratime" => time(),
                );

                $this->db->insert(self::TBL_MSG_RUN, $data);
                $run_id = $this->db->insert_id();

                $mns = $this->get_instance_of_mnslib();
                $mns->send(self::QNAME_MSG_ATOM, $run_id, $delay_seconds);
                @msgtasklog("subdivide_task", func_get_args(), "run_id:{$run_id}");

                $delay_seconds += $sub_typ_conf["interval"];
            }
        }


        return true;
    }


    public function clean_up_tasks()
    {
        $query_task = $this->db->where("stat", self::TSK_STAT_ING)->get(self::TBL_MSG_TASK);

        if (!$query_task) {
            return true;
        }

        $running_tasks = $query_task->result_array();

        foreach ($running_tasks as $msg_task) {
            $query = $this->db->select('rt_id,rstat')->where(
                array("rt_id" => $msg_task["mt_id"], "rstat<=" => self::TSK_STAT_DONE))->get(self::TBL_MSG_RUN);

            if ($query && $query->num_rows()) {
                $data = $query->result_array();

                if (empty($data)) {
                    // 可能任务还未开始, 等待
                    continue;
                }

                foreach ($data as $k => $v) {
                    if ($v['rstat'] == self::TSK_STAT_DONE) {
                        unset($data[$k]);
                    }
                }

                if (empty($data)) {
                    $this->update_task_status($msg_task["mt_id"], self::TSK_STAT_DONE);
                }
            }
        }
        return true;
    }


    public function atom_run_pm_ry_ids($from_uid, $message, $to, $custom_type = 0, $msg_source = 0)
    {
        $this->load->model('imapim');

        $to_arr = explode(self::CHAR_SEP, $to);

        $res = $this->imapim->gs_send_normal_chat($from_uid, $message, $to_arr, $custom_type, $msg_source);
        @msgtasklog('atom_run_pm_ry_ids', func_get_args(), $res);

        $num_all = count($to_arr);
        $num_err = 0;
        if ($res[0]) {  // error
            $num_err = $num_all;
        }
        return [$num_all, $num_err];
    }


    public function atom_run_pm_jmessage_ids($from_uid, $message, $to)
    {
        $this->load->model('imapim');

        $num_err = 0;
        $to_arr = explode(self::CHAR_SEP, $to);
        foreach ($to_arr as $to_id) {
            $res = $this->imapim->jg_im_send_message($to_id, $message, $from_uid);
            $param = array("to_id" => $to_id, "message" => $message, "from_uid" => $from_uid);
            @msgtasklog('atom_run_pm_ry_ids', $param, $res);
            
            if ($res[0]) {
                $num_err += 1;
            }
        }
        
        return [count($to_arr), $num_err];
    }


    public function getMsgAtom($run_id)
    {
        $sql = 'select rt_id,rsub_typ,target_group_typ,b.to as `to`,ratime,run_id,from_id,from_name,message,custom_type,msg_source from ml_msg_task, ml_msg_task_run as b where mt_id=rt_id and run_id=' . $run_id;

        $msg_atom = $this->db->query($sql)->row_array();

        return $msg_atom;
    }

    /**
     *
     * 测试数据:
     * $msg_atom = array(
     * "rt_id" => 2,
     * "target_group_typ" => 0,
     * "rsub_typ" => 1,
     * "to" => 295364,
     * "ratime" => 1461569066,
     * "run_id" => 3,
     * "from_id" => 50,
     * "from_name" => "rongxiang",
     * "message" => 9854,
     * "custom_type"=> 1
     * );
     *
     *
     * @return array|bool
     */
    public function atom_run()
    {
        $mns = $this->get_instance_of_mnslib();
        $time_out = time() + 10 * 60;
        while (1) {
            $msg = $mns->receive(self::QNAME_MSG_ATOM);     // 如果需要, 可以设置消息在获取时不自动删除, 确认被处理再删
            if ($msg[0] || ! $msg[1]) {
                // 暂无消息
                return $msg;
            }

            $run_id = $msg[1];
            $msg_atom = $this->getMsgAtom($run_id);

            $sub_typ = $msg_atom["rsub_typ"];
            $target_group_typ = $msg_atom["target_group_typ"];

            $sub_typ_conf = isset(self::$sub_method_conf[$sub_typ]) ? self::$sub_method_conf[$sub_typ] : null;
            if (empty($sub_typ_conf)) {
                return true;
            }

            if ($target_group_typ == self::TGT_TYP_GROUP && !$sub_typ_conf["is_support_group"]) {
                return true;
            } elseif ($target_group_typ == self::TGT_TYP_ALL && !$sub_typ_conf["is_support_all"]) {
                return true;
            }


            $result = array();
            if ($msg_atom["rsub_typ"] == self::SUB_TYP_PM_RY && $msg_atom["target_group_typ"] == self::TGT_TYP_IDS) {
                $result = $this->atom_run_pm_ry_ids($msg_atom["from_id"], $msg_atom["message"], $msg_atom["to"], $msg_atom["custom_type"], $msg_atom['msg_source']);
            } elseif ($msg_atom["rsub_typ"] == self::SUB_TYP_PM_JMESSAGE && $msg_atom["target_group_typ"] == self::TGT_TYP_IDS) {
                $result = $this->atom_run_pm_jmessage_ids($msg_atom["from_id"], $msg_atom["message"], $msg_atom["to"]);
            }

            $updata = array(
                "num_all" => isset($result[0]) ? $result[0] : 0,
                "num_error" => isset($result[1]) ? $result[1] : 0,
                "result_id" => isset($result[2]) ? $result[2] : '',
                "rstat" => self::TSK_STAT_DONE,
            );
            $this->update_msg_run($msg_atom["run_id"], $updata);

            if (time() > $time_out) {
                break;
            }
        }
        return true;
    }

}